package com.atguigu.state;

import com.atguigu.pojo.WaterSensor;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

/**
 *
 * 案例需求：针对每种传感器输出最高的3个水位值
 *
 * 思路：中间状态可以存多个值，将所有数据存在中间状态中，排序后，去掉第四个后输出
 *
 * 按键分区状态 - 列表状态 - ListState
 *
 *  ListState:  可以维护多个状态值
 *
 *     update(List<T> values) :   将给定的list集合中的数据覆盖掉状态中已有的值
 *     void addAll(List<T> values) : 将给定的list集合中的数据添加到状态中
 *     OUT get():  获取状态中的值
 *     void add(IN value): 将给定的数据添加到状态中
 *     void clear(): 清空状态
 *
 */

public class Flink06_KeyedListState {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // s1,100,1000
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Integer.valueOf(fields[1].trim()), Long.valueOf(fields[2].trim()));
                        }
                );

        ds.print("INPUT");

        ds
                .keyBy(WaterSensor::getId)
                .process(
                        new KeyedProcessFunction<String,WaterSensor, String>() {//TODO 这里三个参分别是什么？

                            //1.声明变量
                            private ListState<Integer> top3VcState;
                            //2.使用生命周期初始化状态


                            @Override
                            public void open(Configuration parameters) throws Exception {
                                RuntimeContext runtimeContext = getRuntimeContext();
                                ListStateDescriptor<Integer> top3VcStateDesc = new ListStateDescriptor<>("listStateDec", Types.INT);
                                top3VcState = runtimeContext.getListState(top3VcStateDesc);

                            }


                            @Override
                            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context context, Collector<String> out) throws Exception {
                                //3.使用状态
                                //将当前的水位值添加到状态中
                                top3VcState.add(value.getVc());

                                List<Integer> top3VcList = new ArrayList<>();
                                top3VcState.get().forEach(top3VcList :: add); //这里不太懂？

                                //排序
                                top3VcList.sort(
                                        new Comparator<Integer>() {
                                            @Override
                                            public int compare(Integer o1, Integer o2) {
                                                return Integer.compare(o2,o1); //倒序排
                                            }
                                        }
                                );

                                //取出状态的值,前三个
                                if (top3VcList.size()>3){
                                    //移除第四个，留3个
                                    top3VcList.remove(3);
                                }
                                //更新状态
                                top3VcState.update( top3VcList );
                            }

                        }
                )
                .print("PROCESS");



        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
